/*
 *  VmGroup.h
 *  vorpal
 *
 *  Created by Russell Y Webb on 24/08/09.
 */

#ifndef VORPAL_VM_GROUP_H
#define VORPAL_VM_GROUP_H

#include "Common.h"
#include "VM.h"
#if useThreads
#include "TThreadPool.hh"
#endif

/// Everything internal to vorpal is inside this namespace.
namespace Vorpal{

	typedef bool (*ReceiveHook)(Message msg, uint32_t numArgs, VM *vm);
	
#if useThreads
	class VmJob : public TThreadPool::TJob{
	public:
		VmJob(){ }
		
		void Attach(VM *vm){ vms.push_back(vm); }
		
		virtual void run(void *custom){
			loop(i, 0, vms.size()){
				uint initialSize = vms[i]->IncomingSize();
				loop(j, 0, initialSize){ // once for each event
					if(vms[i]->ReceivedKill()){ break; }
					vms[i]->NextCall();
				}
			}
		}
	private:
		vector< VM* > vms;
	};
#endif
	
	class VmGroup{
	public:
		// TODO: should not be called, make into singleton?
		VmGroup() : nextNumber(0){ }

		~VmGroup(){
			for(set< VM* >::iterator it = spawnedVM.begin(); it != spawnedVM.end(); ++it){
				delete *it;
			}
		}
		
		string Register(VM *vm){
			string name = ReserveName();
			Register(vm, name);
			return name;
		}
		
		void Remove(VM *vm){
			map< VM*, string >::const_iterator it = vmToName.find(vm);
			if(it != vmToName.end()){
				nameToVm.erase(it->second);
				vmToName.erase(vm);
			}
			if(spawnedVM.find(vm) != spawnedVM.end()){
				spawnedVM.erase(vm);
				delete vm;
			}
		}
		
		VM *operator[](const string &name){
			map< string, VM* >::iterator found = nameToVm.find(name);
			if(found == nameToVm.end()){
				return 0;
			}
			return found->second;
		}
		
		const string &operator[](VM *vm){
			return vmToName[vm];
		}
		
		bool RoundRobin(){
			int activityCount = 0;
			for(map< VM*, string >::const_iterator it = vmToName.begin(); it != vmToName.end(); ++it){
				activityCount += it->first->SendCalls();
				activityCount += it->first->UpdateState();
				activityCount += it->first->IncomingCount();
			}			
#if useThreads
			if(traceProcess){ cout << "{"; }
			uint numThreads = tp_max_parallel();
			DeclareArray(job, VmJob, numThreads);
			
			uint i = random();
			for(map< VM*, string >::const_iterator it = vmToName.begin(); it != vmToName.end(); ++it){
				job[i % numThreads].Attach(it->first);
				i += 1;
			}
			loop(i, 0, numThreads){
				tp_run(&job[i]);
			}
			tp_sync_all();					
			if(traceProcess){ cout << "}\n"; }
#else
			for(map< VM*, string >::const_iterator it = vmToName.begin(); it != vmToName.end(); ++it){
				loop(i, 0, it->first->IncomingSize()){
					it->first->NextCall();
				}
			}
#endif
			for(map< VM*, string >::const_iterator it = vmToName.begin(); it != vmToName.end(); ++it){
				if(it->first->ReceivedKill()){
					Remove(it->first);
					it = vmToName.begin(); // back to beginning
				}
			}
			//cout << "active: " << activityCount << endl;
			return activityCount > 0;
		}
		
		void Show(){
			for(map< VM*, string >::const_iterator it = vmToName.begin(); it != vmToName.end(); ++it){
				cout << it->first << " " << it->second << endl;
			}			
			for(map< string, VM* >::const_iterator it = nameToVm.begin(); it != nameToVm.end(); ++it){
				cout << it->first << " " << it->second << endl;
			}			
		}
		
		string ReserveName(){
			ostringstream os;
			os << "process-" << nextNumber++;
			return os.str();
		}
		
		void Receive(const RemoteCall &rc){
			switch(rc.GetMessage()){
				case MSG_spawn: {
					// create vm
					Handle<Object> newSys;
					VM *newVm = new VM(newSys, false); 
					spawnedVM.insert(newVm);
					// create system
					istringstream is(rc[0]);
					newSys = Object::Unserialize(is, newVm);
					newVm->SetSystem(newSys);
					
					Register(newVm, rc.Destination());
					break;
				}
				case MSG_kill: {
					Remove((*this)[rc.Destination()]);
					break;
				}
				default:
					Failure(ERR_INVALID_MESSAGE_PASSED);
			}
		}
		
		void AddReceiveHook(ReceiveHook hook){
			hooks.push_back(hook);
		}
		
		bool Receive(Message msg, uint32_t numArgs, VM *vm){
			loop(i, 0, hooks.size()){
				if(hooks[i](msg, numArgs, vm)){
					return true;
				}
			}
			return false;
		}
		
	private:
		void Register(VM *vm, const string &name){
			map< VM*, string >::const_iterator it = vmToName.find(vm);
			Assert(it == vmToName.end()); // added twice
			vmToName[vm] = name;
			nameToVm[name] = vm;
			//cout << "CREATING PROCESS " << name << " " << vm << endl;
		}
				
		uint nextNumber;
		map< VM*, string > vmToName;
		map< string, VM* > nameToVm;
		set< VM* > spawnedVM;
		vector< ReceiveHook > hooks;
	};

	VmGroup &TheGroup();
	uint GlobalUniqueId();
}

#endif
